package com.ustcinfo.study.worker1.r5

import org.apache.spark.{SparkConf, SparkContext}

/**
  * Auther: pengqiang
  * Date: 2018/8/4 12:53
  * Description: r5 work
  */
object CityDnsTopN {
  /**
    * ip转换为Long类型方法
    * @param ip 127.0.0.1
    * @return long类型
    */
  def ip2Long(ip: String): Long = {
    val arr = ip.split("[.]")
    var num = 0L
    for (i <- 0 until arr.length) {  //until不包含上限，如：1 to 3, Range为1,2
      num = arr(i).toLong | num << 8L
    }
    num
  }

  /**
    * 二分查找思想
    * @param arr  定位库
    * @param ip  需要定位的IP
    * @return 数组下标
    */
                              //下限， 上限    城市   需要定位的ip
  def binarySearch(arr: Array[(Long, Long, String)], ip: Long): Int = {
    var l = 0
    var h = arr.length - 1  //定位库的总长度
    while (l <= h) {  //遍历定位库
      var m = (l + h) / 2
      if ((ip >= arr(m)._1) && (ip <= arr(m)._2)) {  //如果需要定位的IP在该城市范围之类，返回城市所在的数组下标
        return m
      } else if (ip < arr(m)._1.toLong) {  //如果小于下限，设上限为该下标-1
        h = m - 1
      } else {                       //如果大于上限 ，设下限为该下标+1
        l = m + 1
      }
    }   //遍历结束
    -1  //定位失败返回-1
  }

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("pengqiang_scala_test")
    val sc = new SparkContext(sparkConf)
    // 创建sqlContext 并将sparkContext加载到SQLContext中
//    val sqlContext = new SQLContext(sc)

//    val hiveContext = new HiveContext(sc)

    val cityRdd = sc.textFile("src/main/resources/city_dns.txt")  //定位文件
      .filter(x => x.trim.length > 0)   //过滤空行，降低噪声
      .map(x => {
        val s = x.split("\t")
        (ip2Long(s(0)),ip2Long(s(1)),s(2))   //映射为指定格式
      }).sortBy(_._1)  //下限，上限，城市   //根据下限排序

    val bd = sc.broadcast(cityRdd.collect)    //设置广播变量，让该rdd全结点可用

    val dnsRdd = sc.textFile("src/main/resources/data/dns_test.txt")  //输入日志文件1
      .filter(x => x.trim.length > 0)   //过滤空行，降低噪声
      .map(x => {
      val p = x.split("\\|")          //这里为|不报错，但是分隔不出来，一个坑点，需要转义
      (p(0),p(1)) //数据集为 IP  ，域名
    }).sortBy(_._1)                 //根据ip进行排序

    val dns1Rdd = sc.textFile("src/main/resources/data/dns_test1.txt") //输入日志文件2
      .filter(x => x.trim.length > 0)   //过滤空行，降低噪声
      .map(x => {
      val p = x.split("\\|")          //这里为|不报错，但是分隔不出来，一个坑点，需要转义
      (p(0),p(1)) //数据集为 IP  ，域名
    }).sortBy(_._1)                 //根据ip进行排序

//    dnsRdd.union(dns1Rdd)  //此方法rdd加和失败，原因未知
    val resultRdd = dnsRdd union dns1Rdd

    resultRdd.map(x => {
      val ipLong = ip2Long(x._1)  //获取ip的long类型
      //获取下标
      val index = binarySearch(bd.value, ipLong)
      //没找到的返回unknown
      if (index == -1) {
        (x._1, x._2, "unknown")
      } else {
        //获取城市
        val p = bd.value(index)._3
        ( x._1, x._2, p)  // ip ，域名 ， 城市
      }
    })

      resultRdd.map(x=> x.toString()).map(x => x.substring(1,x.length-1)) //将输出内容为(ip,域名,城市)的二维数组
                                                                          //map为 ip,域名,城市 的字符串数组
//    --------------------------------------------------------------------------
      resultRdd.repartition(1).saveAsTextFile("E:\\outputFile")   //输出外部文件内容为 ip,域名,城市
                                                                                  //    ip,域名,城市

    sc.stop()//程序结束
  }
}
